package cn.jly.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author lanyangji
 * @date 2019/12/4 10:33
 */
object SparkStreaming02_FileStream {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming02_FileStream")
    val streamingContext = new StreamingContext(sparkConf, Seconds(5))

    // 从指定文件夹中采集数据
    // 监控文件夹创建 DStream
    // 这是监控本地文件系统
    // 也可以监控HDFS上的目录
    // 注意：这种方式很少用，因为有更专业的flume去采集，只要把flume采集到的数据给spark处理就可以了
    val dirStream: DStream[String] = streamingContext.textFileStream("file:///d:/fileStream")

    // 数据切分，形成一个个单词
    val wordsStream: DStream[String] = dirStream.flatMap(_.split("\t"))

    // 映射成元组 (word,1)
    val wordAndOneStream: DStream[(String, Int)] = wordsStream.map((_, 1))

    // 统计
    val wordAndCountStream: DStream[(String, Int)] = wordAndOneStream.reduceByKey(_+_)

    // 打印
    wordAndCountStream.print()

    // 启动spark StreamingContext
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
